package org.yagnus.yadoop

import java.io.DataInput
import java.io.DataOutput
import java.io.IOException
import org.apache.hadoop.io._;
import scala.collection.mutable.ArrayBuffer

@inline class LongArrayWritable extends WritableComparable[LongArrayWritable] {
  var array: Array[Long] = _;

  var equalLengthOnly = true;

  def this(in: Array[Long]) {
    this();
    set(in);
  }
  def this(in: Long*) {
    this();
    set(in.toArray);
  }

  def set(values: Array[Long]) = {
    this.array = values;
  }

  def get: Array[Long] = array;
  def length: Long = {
    if (array == null) return 0;
    else return array.length;
  }
  def apply(ind: Int) = array(ind);

  override def readFields(in: DataInput) {
    val serializer = new VLongWritable();
    serializer.readFields(in);
    array = null;
    val len = serializer.get.intValue;
    val newArray = new Array[Long](len); // construct values
    for (i ← 0 until len) {
      serializer.readFields(in)
      newArray(i) = serializer.get;
    }
    set(newArray);
  }

  override def write(out: DataOutput) {
    val serializer = new VLongWritable();
    if (array == null) {
      serializer.set(0);
      serializer.write(out);
    } else {
      serializer.set(array.length);
      serializer.write(out);
      for (l ← array) {
        serializer.set(l);
        serializer.write(out);
      }
    }
  }

  override def equals(o: Any): Boolean = {
    if (o == null) return false;
    if (!o.isInstanceOf[LongArrayWritable]) return false;
    return compareTo(o.asInstanceOf[LongArrayWritable]) == 0;
  }
  override def compareTo(o: LongArrayWritable): Int = {
    if (o eq null) {
      if (equalLengthOnly) throw new LongArrayWritableError("LongArrayWritable.compaerTo received a null array to compare to.");
      return 1;
    }
    if (array.length == o.array.length) {
      for (i ← 0 until array.length) {
        if (this.array(i) > o.array(i)) {
          return 1
        } else if (this.array(i) < o.array(i)) {
          return -1;
        }
      }
      return 0;
    } else if (array.length > o.array.length) {
      if (equalLengthOnly) throw new LongArrayWritableError("LongArrayWritable.compaerTo received a comparison to a shorter array");
      return 1;
    } else {
      if (equalLengthOnly) throw new LongArrayWritableError("LongArrayWritable.compaerTo received a comparison to a longer array");
      return -1;
    }
  }
  def getLongArrayWritable = this;
  def toLongArrayWritable = this;
  def clear { array = null; }
  def reset = clear;
}

class LongArrayWritableError(x: String) extends Exception(x) {
}

@inline final class AuxLongArrayWritable extends LongArrayWritable {
  import org.yagnus.scalasupport.iterates.BreakableIterator._;

  var dataItr: () => Iterator[Long] = null;
  var len: Long = 0;
  def size = len;

  //If the data was specified, then use the iterator for output
  def this(in: Iterator[Long], len: Long) {
    this();
    this.dataItr = () => in;
    this.len = len;
  }
  def this(in: Iterable[Long], len: Long) {
    this();
    this.dataItr = () => in.iterator;
    this.len = len;
  }

  //if must be the iterator will be made into an array and returned.
  override def get = {
    if (array == null && dataItr != null) {
      array = (new ArrayBuffer() ++= dataItr()).toArray;
      dataItr = () => array.iterator;
    }
    array;
  }
  //if this data was read in, then use that as output.
  override def set(newArr: Array[Long]) {
    super.set(newArr);
    if (newArr != null) {
      this.dataItr =
        if (array != null) () => array.iterator;
        else () => Iterator[Long]();
      len = array.length;
    }
  }
  override def write(out: DataOutput) {
    val serializer = new VLongWritable();
    var limit = len;
    serializer.set(limit);
    serializer.write(out);

    for (l ← dataItr() breakable) {
      if (limit > 0) {
        serializer.set(l);
        serializer.write(out);
        limit -= 1;
      } else l.break;
    }
  }
  override def length = len;
  override def clear { dataItr = null; array = null; len = 0 }
  override def reset = clear;
}

/**
 * The long iterable has dual view of data. It can eithe store data as an array or it
 * is an iterable of longs. This enables us to serialize the input into array when reading
 * but produce data one at a time in an iterable and write them sequentially without requiring
 * the memory for all the output to be allocated. The object can be in one of three states:
 *
 * * uninited: array backing and iterable both null
 * * array inited: array and iterable both intialized, reads into array, writes via iterable
 * * iterable inited: array backing is null, iterable is intialized, reads into array, writes from iterable.
 *
 * reset() returns object to uninited state.
 */
@inline final class LongIterableWritable extends LongArrayWritable with Iterable[Long] {
  import org.yagnus.scalasupport.iterates.BreakableIterator._;

  var data: Iterable[Long] = null;
  def iterator = data.iterator;

  var len: Int = 0;
  override def length = len;

  //If the data was specified, then use the iterator for output
  def this(in: Iterable[Long], len: Long) {
    this();
    this.data = in;
    this.len = len.intValue;
  }

  //if this data was read in, then use that as output.
  override def set(newArr: Array[Long]) {
    super.set(newArr);
    if (array != null) {
      data = array;
      len = array.length;
    } else {
      data = Iterable[Long]();
      len = 0;
    }
  }

  //if must be the iterator will be made into an array and returned.
  override def get = {
    if (array == null && data != null) {
      array = (new ArrayBuffer() ++= data).toArray;
      data = array;
    }
    array;
  }

  //if this data was read in, then use that as output.
  def set(newData: Iterable[Long], size: Long) {
    super.set(null);
    this.data = newData;
    this.len = size.intValue;
  }
  def set(newData: Iterable[Long]) {
    set(newData, newData.size);
  }
  override def write(out: DataOutput) {
    val serializer = new VLongWritable();
    var limit = len;
    serializer.set(limit);
    serializer.write(out);
    for (l ← iterator breakable) {
      if (limit > 0) {
        serializer.set(l);
        serializer.write(out);
        limit -= 1;
      } else l.break;
    }
  }
  override def clear { data = null; array = null; len = 0 }
  override def reset = clear;
}
